package com.zlx.taskcenter.common;

import com.zlx.redismessage.util.RedisUtils;
import com.zlx.taskcenter.consumer.BaseConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.QueryTimeoutException;

import java.util.concurrent.TimeUnit;

/**
 * 队列监听器
 * @author zlx
 */
public class RedisQueueListener implements Runnable {

    public static final Logger log = LoggerFactory.getLogger(RedisQueueListener.class);
    private String queue;
    private BaseConsumer consumer;

    public RedisQueueListener(String queue, BaseConsumer consumer) {
        this.queue = queue;
        this.consumer = consumer;
    }


    @Override
    public void run() {
        log.info("QueueListener start...queue:{}", queue);
        while (RedisConsumerContainer.run) {
            try {
                // 阻塞
//                Object msg = redisTemplate.opsForList().rightPopAndLeftPush(queue, queue + "Bak", 30, TimeUnit.SECONDS);
                Object msg = RedisUtils.getRedisTemplate().opsForList().rightPop(queue, 30, TimeUnit.SECONDS);
                if (msg != null) {
                    try {
                        consumer.onMessage(String.valueOf(msg));
                    } catch (Exception e) {
                        consumer.onError(String.valueOf(msg), e);
                    }
                }
            } catch (QueryTimeoutException ignored) {
            } catch (Exception e) {
                if (RedisConsumerContainer.run) {
                    log.error("Queue:{}", queue, e);
                } else {
                    log.info("QueueListener exits...queue:{}", queue);
                }
            }
        }
    }

}
